package org.datastreamapi.transform

import org.FlinkStreamApp
import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala._

/**
 * description ：多流 connect
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/21 20:17
 * modified By ：
 * version:    : 1.0
 */
object ConnectedStreams extends FlinkStreamApp {

  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 创建数据
    val source1 = environment.fromElements((1, 11), (2, 22))
    val source2 = environment.fromElements((1, "a"), (2, "b"))

    // 多流 connect - 01
    //val value = source1.keyBy(_._1).connect(source2.keyBy(_._1))

    // 多流 connect - 01
    val value = source1.connect(source2).keyBy(0, 0)

    val result = value.map(new MyCoMapFunction)

    // 输出
    result.print
  }

  class MyCoMapFunction extends CoMapFunction[(Int, Int), (Int, String), String] {
    override def map1(in1: (Int, Int)) = in1._2 + " - from map1"

    override def map2(in2: (Int, String)) = in2._2 + " - from map2"
  }

}
